package com.nightsoul.hadoop1.test.sort;

import java.io.IOException;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nightsoul.hadoop1.test.OldJobBuilder;
import com.nightsoul.hadoop1.test.junit.NcdcRecordParser;

/**
 * 将气象数据转换为SequenceFile，并没有进行排序
 * @author zj
 *
 */
public class SortDataProgressor extends Configured implements Tool {
	static class CleanerMapper extends MapReduceBase
		implements Mapper<LongWritable, Text, IntWritable, Text> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
				throws IOException {
			parser.parse(value);
			if(parser.isValidTemperature()) {
				output.collect(new IntWritable(parser.getAirTemperature()), value);
			}
			
		}
		
	}
	
	
	@Override
	public int run(String[] args) throws Exception {
		JobConf conf = OldJobBuilder.parseInputAndOutput(this, getConf(), args);
		if(conf==null) {
			return -1;
		}
		
		conf.setMapperClass(CleanerMapper.class);
		conf.setOutputKeyClass(IntWritable.class);
		conf.setOutputValueClass(Text.class);
		conf.setNumReduceTasks(0);
		conf.setOutputFormat(SequenceFileOutputFormat.class);
		SequenceFileOutputFormat.setCompressOutput(conf, true);
		SequenceFileOutputFormat.setOutputCompressorClass(conf, GzipCodec.class);
		SequenceFileOutputFormat.setOutputCompressionType(conf, CompressionType.BLOCK);
		
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new SortDataProgressor(), args);
		System.exit(exitCode);
	}
}
